package mapred.example;

import java.io.IOException;

import mapred.base.TextMapper;
import mapred.example.model.Link;
import mapred.example.model.Page;
import mapred.example.model.Paragraph;

import org.apache.hadoop.io.Text;

public abstract class ParseReader extends TextMapper {

    private static final String START_SIGN = "#####START_SIGN#####";

    private static final String PAGE_URL = "#####PAGE_URL#####";

    private static final String NORMALIZED_PAGE_URL = "#####NORMALIZED_PAGE_URL#####";

    private static final String SEGMENTED_PAGE_URL = "#####SEGMENTED_PAGEURL######";

    private static final String TITLE = "#####SEGMENTED_TITLE#####";

    private static final String OUT_LINKS = "#####OUT_LINKS#####";

    private static final String BODY = "#####SEGMENTED_PAGE_BODY#####";

    private static final String FINISH_SIGN = "#####FINISH_SIGN#####";

    private Page page = null;

    private String last = "";

    private int outdegree = 0;

    @Override
    public void map(Text key, Text value) throws IOException {
        String k = key.toString();
        if (START_SIGN.equals(k)) {
            page = new Page();
        } else if (FINISH_SIGN.equals(k)) {
            map(page);
        } else if (PAGE_URL.equals(last)) {
            page.getUrl().setRaw(k);
        } else if (NORMALIZED_PAGE_URL.equals(last)) {
            page.getUrl().setNormalized(k);
        } else if (SEGMENTED_PAGE_URL.equals(last)) {
            page.getUrl().setText(combine(k, value.toString()));
        } else if (TITLE.equals(last)) {
            page.setTitle(combine(k, value.toString()));
        } else if (OUT_LINKS.equals(last)) {
            outdegree = Integer.parseInt(k);
        } else if (outdegree > 0) {
            try {
                page.addLink(Link.newInstanceFromParser(combine(k, value
                        .toString())));
                this.incrCounter("Links", "Good", 1L);
            } catch (Exception e) {
                this.incrCounter("Links", "Error", 1L);
            }
            outdegree--;
        } else if (BODY.equals(last)) {
            String body = combine(k, value.toString());
            for (String p : body.split("\1")) {
                page.addParagraph(Paragraph.newInstanceFromParser(p));
            }
        }
        last = k;
    }

    private String combine(String k, String v) {
        if (v.isEmpty()) {
            return k;
        } else {
            return k + "\t" + v;
        }
    }

    public abstract void map(Page page) throws IOException;

}
